package _50_rocketmq._producer01;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * send( , sendCallback)
 *      获取 发送结果，根据 是否成功，单独处理
 *      注：想要拿到 返回结果，不能 shutdown
 */
public class ProducerSendAsyncTest {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        // 使用默认类，创建一个 生产者实例，并配置参数
        DefaultMQProducer producer = new DefaultMQProducer("test_aa_group_name");
        producer.setNamesrvAddr("localhost:9876");
        // 设置超时时间
        producer.setSendMsgTimeout(5000);
        // 启动 生产者 服务
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(3);

        // 由于是异步发送，引入 countDownLathc，保证所有生产者发送消息的回调函数 执行完毕 再停止 producer 服务
        int msgCount = 5;
        CountDownLatch countDownLatch = new CountDownLatch(5);

        // 发送多条消息
        for (int i = 0; i < msgCount; i++) {
            try {
                final int index = i;
                // 创建消息，并配置 topic 等参数
                Message message = new Message(
                        "TopicTest",
                        "TagA",
                        "hello mq".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 生产者 发送消息，对发送成功和失败 单独处理，需要拿到 返回结果，需保持 start 状态
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d Exception %s %n", index, throwable);
                        throwable.printStackTrace();
                    }
                });
                System.out.println("消息发送完成");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        // 关闭 生产者 服务
        System.out.println("等待回调...");
        countDownLatch.await(10, TimeUnit.SECONDS);
        System.out.println("关闭服务");
        producer.shutdown();
    }

}
